package com.sduept.component.message.core;

import com.sduept.bigdata.fault.facde.FaultGenerateFacde;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class CentralizedWaveAnalysisOverMessageService {
	@Autowired
	private FaultGenerateFacde facde;
	private ExecutorService executor = Executors.newFixedThreadPool(5);

	@KafkaListener(topics = { "lbq_interaction_result" })
	public void onMessage(ConsumerRecord<?, ?> record) {
		Optional<?> kafkaMessage = Optional.ofNullable(record.value());
		if (kafkaMessage.isPresent()) {
			JSONObject jvalue = JSONObject.fromObject(JSONObject.fromObject(kafkaMessage.get()).getString("value"));
			String remoteFileUUID = jvalue.getString("remoteFileUUID");
			String result = jvalue.getString("result");
			log.info("接收到消息：lbq_interaction_result, remoteFileUUID表：" + remoteFileUUID + ",result：" + result);
			if ("0".equals(result)) {
				final String faultFileId = jvalue.getString("localFileUUID");
				executor.submit(new Runnable() {
					@Override
					public void run() {
						if (StringUtils.isEmpty(remoteFileUUID)) {
							if (!StringUtils.isEmpty(faultFileId)) {
								facde.generateFaultByCWave(faultFileId);
							}
						} else {
							facde.generateFaultByCWave(faultFileId);
						}
					}
				});
			}
		}

	}
}
